package com.markhsiu.store.disk;

import java.io.File;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import com.markhsiu.store.StoreDB;
import com.markhsiu.store.StoreService;

import io.openmessaging.demo.DefaultBytesMessage;

public class StoreTopicService implements StoreService {

	private static final Object EMPTY = new Object();
	private static final Map<String, TopicSegment> topicProducer = new ConcurrentHashMap<>(156);
	private static final Map<String, Object> topicSet = new ConcurrentHashMap<>(156);

	private static final Map<String, TopicSegment> topicConsumer = new HashMap<>(156);
	
	@Override
	public void init() {
		if (StoreDB.isProducer) {
			return;
		}

		scanTopicDir();
	}

	private void scanTopicDir() {

		File pathDir = new File(StoreDB.getRootDir());
		File[] topicRootDir = pathDir.listFiles();
		for (File topicFile : topicRootDir) {
			String topic = topicFile.getName();
			TopicSegment queueDisk = new TopicSegment();
			queueDisk.load(topicFile);
			topicConsumer.put(topic, queueDisk);
		}
	}

	@Override
	public void put(DefaultBytesMessage message) {
		String topic = message.getTopic();
//		long s1 = System.nanoTime();
		TopicSegment queueDisk = topicProducer.get(topic);
//		long s2 = System.nanoTime();
//		time.addAndGet(s1-s2);
		
		if (queueDisk == null) {
			// 原子性方法
			// 如果（调用该方法时）key-value 已经存在，则返回那个 value 值。如果调用时 map 里没有找到 key 的
			// mapping，返回一个 null 值
			Object topicLock = topicSet.putIfAbsent(topic, EMPTY);
			if (topicLock == null) {
				queueDisk = new TopicSegment();
				queueDisk.builder(topic);
				topicProducer.put(topic, queueDisk);

			} else {
				try {
					while (!topicProducer.containsKey(topic)) {
						TimeUnit.MILLISECONDS.sleep(1);
						continue;
					}
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				queueDisk = topicProducer.get(topic);
				
			}
		}
		
		queueDisk.putBuffM(message);
		 
	}

	@Override
	public DefaultBytesMessage poll(String topic) {

		Cache cache = topicConsumer.get(topic).getCacheM();
		if (cache == null) {
			return null;
		}
	
		DefaultBytesMessage message = StoreDB.messagePool.get();
		if (message == null) {
			message = new DefaultBytesMessage();
			StoreDB.messagePool.set(message);
		}
		
		int length = MessageUtil.restore(cache.buff, cache.offset, message);
		cache.offset += length;
		 
		return message;
	}

	@Override
	public void flush() {
		
		for (TopicSegment segment : topicProducer.values()) {
			segment.flush();
		}
	

	
//		System.out.println(Thread.currentThread().getName() + "  Service "+(time.get()/1000000) +" ms");
	}

	@Override
	public void destory() {}

}
